package com.winning.devops.demo.base.queue;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import com.winning.devops.demo.base.disruptor.LoggerEvent;
import com.winning.devops.demo.base.disruptor.LoggerEventFactory;
import com.winning.devops.demo.base.disruptor.LoggerEventHandler;
import com.winning.devops.demo.base.model.LoggerMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author chensj
 * @title 创建Disruptor日志队列
 * @project logback-demo
 * @package com.winning.devops.demo.base.queue
 * @date: 2019-04-09 16:26
 */
@Component
public class DisruptorLoggerQueue {
    private static final Logger LOG = LoggerFactory.getLogger(DisruptorLoggerQueue.class);
    /**
     * RingBuffer 大小，必须是 2 的 N 次方；
     */
    int ringBufferSize = 1024 * 1024;
    /**
     * 构造器初始化Disruptor
     */
    private Disruptor<LoggerEvent> disruptor = new Disruptor<>(LoggerEvent::new, ringBufferSize, DaemonThreadFactory.INSTANCE);
    /**
     * 静态变量 环形的缓冲区
     * 负责对通过 Disruptor 进行交换的数据（事件）进行存储和更新
     */
    private static RingBuffer<LoggerEvent> ringBuffer;

    @Autowired
    public DisruptorLoggerQueue(LoggerEventHandler eventHandler) {
        LOG.info(getClass() + " construct init");
        System.out.println("init DisruptorLoggerQueue ");
        disruptor.handleEventsWith(eventHandler);
        ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
    }

    /**
     * 事件发布
     *
     * @param log
     */
    public static void publishEvent(LoggerMessage log) {
        // 请求下一个事件序号；
        long sequence = ringBuffer.next();
        try {
            // //获取该序号对应的事件对象；
            LoggerEvent event = ringBuffer.get(sequence);
            // 设置事件
            event.setLogger(log);
        } finally {
            // 发布事件；
            ringBuffer.publish(sequence);
        }
    }
}
